package com.bomb.disruptor.multi;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;

/**
 * 多生产者，多消费者
 */
public class Main {
	
	public static void main(String[] args) throws Exception {
		// 创建ringBuffer
		RingBuffer<Order> ringBuffer = 
				RingBuffer.create(ProducerType.MULTI, 
						new EventFactory<Order>() {  
				            @Override  
				            public Order newInstance() {  
				                return new Order();  
				            }  
				        }, 
				        1024 * 1024, 
						new YieldingWaitStrategy());
		

		// 多个消费者
		Consumer[] consumers = new Consumer[10];
		for(int i = 0; i < consumers.length; i++){
			consumers[i] = new Consumer("consumer" + i);
		}
		WorkerPool<Order> workerPool =
				new WorkerPool<Order>(ringBuffer,
						ringBuffer.newBarrier(),
						new IntEventExceptionHandler(),
						consumers);
        ringBuffer.addGatingSequences(workerPool.getWorkerSequences());

		// 多线程启动
		workerPool.start(Executors.newFixedThreadPool(20));

		// 生产者生产数据
        final CountDownLatch latch = new CountDownLatch(1);
        for (int i = 0; i < 100; i++) {  
        	final Producer p = new Producer(ringBuffer);
        	new Thread(new Runnable() {
				@Override
				public void run() {
					try {
						// 所有的一起阻塞
						latch.await();
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					// 同时生产数据
					for(int j = 0; j < 100; j ++){
						p.onData(UUID.randomUUID().toString());
					}
				}
			}).start();
        } 
        Thread.sleep(2000);
        System.out.println("---------------开始生产-----------------");
        latch.countDown();
        Thread.sleep(5000);
        System.out.println("总数:" + consumers[0].getCount() );
	}
	
	static class IntEventExceptionHandler implements ExceptionHandler {  
	    public void handleEventException(Throwable ex, long sequence, Object event) {}  
	    public void handleOnStartException(Throwable ex) {}  
	    public void handleOnShutdownException(Throwable ex) {}  
	} 
}
